package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.time.Duration;

/**
 * @author hangyangfei
 * 交易域订单预处理表
 * 需要启动的进程：zk、kafka、maxwell、DwdTradeOrderPreProcess
 */

public class DwdTradeOrderPreProcess {
    public static void main(String[] args)  {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(905L));

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取kafka_db中的数据
        String topicName = "topic_db";
        String groupID = "dwd_trade_order_pre_process";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` bigint,\n" +
                "  `xid` bigint,\n" +
                "  `commit` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string>, \n" +
                "   pt AS PROCTIME() \n" +
                ") " + KafkaUtil.getKafkaDDL(topicName, groupID));

        // TODO 4 过滤出订单详情表
        Table orderDetailTable = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['course_id'] course_id,\n" +
                "  `data`['course_name'] course_name,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `pt`,\n" +
                "  `ts`\n" +
                "from topic_db" );
//                "where `table`='order_detail'\n" +
//                "and `type`='insert'");
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        // TODO 5 过滤出订单表
        Table orderInfoTable = tableEnv.sqlQuery("select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['user_id'] user_id,\n" +
                "   `data`['order_status'] order_status,\n" +
                "   `data`['province_id'] province_id,\n" +
                "   `data`['session_id'] session_id,\n" +
                "   `data`['expire_time'] expire_time,\n" +
                "   `type`,\n" +
                "   `old`,\n" +
                "   `pt`,\n" +
                "   `ts`\n" +
                "from topic_db\n" +
                "where `table`='order_info'\n" +
                "and (`type`='insert' or `type`='update')");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        // TODO 7 join读取的2张表格
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "   od.id,\n" +
                "   od.order_id,\n" +
                "   od.course_id,\n" +
                "   od.course_name,\n" +
                "   od.user_id,\n" +
                "   od.origin_amount,\n" +
                "   od.coupon_reduce,\n" +
                "   od.final_amount,\n" +
                "   od.create_time,\n" +
                "   date_format(od.create_time,'yyyy-MM-dd') date_id,\n" +
                "   od.ts od_ts,\n" +
                "   oi.order_status,\n" +
                "   oi.province_id,\n" +
                "   oi.session_id,\n" +
                "   oi.expire_time,\n" +
                "   date_format(oi.expire_time,'yyyy-MM-dd') expire_time_date,\n" +
                "   oi.ts oi_ts,\n" +
                "   oi.pt,\n" +
                "   oi.`type`,\n" +
                "   oi.`old`,\n" +
                "   current_row_timestamp() row_op_ts\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id");
        tableEnv.createTemporaryView("result_table",resultTable);

        // TODO 8 写入到kafka新的主题中dwd_trade_order_pre_process
        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "   id string,\n" +
                "   order_id string,\n" +
                "   course_id string,\n" +
                "   course_name string,\n" +
                "   user_id string,\n" +
                "   origin_amount string,\n" +
                "   coupon_reduce string,\n" +
                "   final_amount string,\n" +
                "   create_time string,\n" +
                "   date_id string,\n" +
                "   od_ts bigint,\n" +
                "   order_status string,\n" +
                "   province_id string,\n" +
                "   session_id string,\n" +
                "   expire_time string,\n" +
                "   expire_time_date string,\n" +
                "   oi_ts bigint,\n" +
                "   pt TIMESTAMP_LTZ(3),\n" +
                "   `type` string,\n" +
                "   `old` map<string,string>,\n" +
                "   row_op_ts TIMESTAMP_LTZ(3),\n" +
                "   PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaSinkDDL("dwd_trade_order_pre_process"));

        tableEnv.executeSql("insert into dwd_trade_order_pre_process select * from result_table");
    }
}
